/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.broker.longpolling;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.broker.BrokerController;
import com.alibaba.rocketmq.common.ServiceThread;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;

/**
 * 拉消息请求管理，如果拉不到消息，则在这里Hold住，等待消息到来
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-26
 */
public class PullRequestHoldService extends ServiceThread {
	private static final Logger log = LoggerFactory.getLogger(LoggerName.BrokerLoggerName);
	private static final String TOPIC_QUEUEID_SEPARATOR = "@";

	private ConcurrentHashMap<String/* topic@queueid */, ManyPullRequest> pullRequestTable = new ConcurrentHashMap<String, ManyPullRequest>(1024);

	private final BrokerController brokerController;

	public PullRequestHoldService(final BrokerController brokerController) {
		this.brokerController = brokerController;
	}

	private String buildKey(final String topic, final int queueId) {
		StringBuilder sb = new StringBuilder();
		sb.append(topic);
		sb.append(TOPIC_QUEUEID_SEPARATOR);
		sb.append(queueId);
		return sb.toString();
	}

	public void suspendPullRequest(final String topic, final int queueId, final PullRequest pullRequest) {
		String key = this.buildKey(topic, queueId);
		ManyPullRequest mpr = this.pullRequestTable.get(key);
		if (null == mpr) {
			mpr = new ManyPullRequest();
			ManyPullRequest prev = this.pullRequestTable.putIfAbsent(key, mpr);
			if (prev != null) {
				mpr = prev;
			}
		}

		mpr.addPullRequest(pullRequest);
	}

	private void checkHoldRequest() {
		// 遍历PullRequestHoldService.pullRequestTable: ConcurrentHashMap<String/* topic@queueid */,ManyPullRequest/*一个封装ArrayList<PullRequest>列表的对象*/>变量中的每个Key值
		for (String key : this.pullRequestTable.keySet()) {
			String[] kArray = key.split(TOPIC_QUEUEID_SEPARATOR);
			if (kArray != null && 2 == kArray.length) {
				String topic = kArray[0];
				int queueId = Integer.parseInt(kArray[1]);
				// 获取该topic和queueId下面的队列的最大逻辑Offset
				final long offset = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, queueId);
				// 主要的拉取消息逻辑在此方法中
				this.notifyMessageArriving(topic, queueId, offset);
			}
		}
	}

	public void notifyMessageArriving(final String topic, final int queueId, final long maxOffset) {
		String key = this.buildKey(topic, queueId);
		// ManyPullRequest对象中包含了PullRequest的集合列表；
		ManyPullRequest mpr = this.pullRequestTable.get(key);
		if (mpr != null) {
			List<PullRequest> requestList = mpr.cloneListAndClear();
			if (requestList != null) {
				List<PullRequest> replayList = new ArrayList<PullRequest>();

				for (PullRequest request : requestList) {
					// 查看是否offset OK
					// 2.2检查当前最大逻辑offset（入参maxOffset）是否大于该PullRequest对象的pullFromThisOffset值（该值等于当初请求消息的queueOffset值）；
					if (maxOffset > request.getPullFromThisOffset()) {
						try {
							// 若当前最大逻辑offset大于请求的queueOffset值，======>说明请求的读取偏移量已经有数据达到了
							// 进行消息的拉取
							this.brokerController.getPullMessageProcessor().excuteRequestWhenWakeup(request.getClientChannel(), request.getRequestCommand());
						} catch (RemotingCommandException e) {
							log.error("", e);
						}
						continue;
					}
					// 尝试取最新Offset
					else {
						// 若当前最大逻辑offset小于请求的queueOffset值
						final long newestOffset = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, queueId);
						// 则再次读取最大逻辑offset进行第2.2步的尝试(重复else前面的逻辑)
						if (newestOffset > request.getPullFromThisOffset()) {
							try {
								this.brokerController.getPullMessageProcessor().excuteRequestWhenWakeup(request.getClientChannel(), request.getRequestCommand());
							} catch (RemotingCommandException e) {
								log.error("", e);
							}
							// 继续遍历下一个PullRequest对象
							continue;
						}
					}

					// 若当前最大逻辑offset还是小于请求的queueOffset值,检查该请求是否超时
					// 查看是否超时
					if (System.currentTimeMillis() >= (request.getSuspendTimestamp() + request.getTimeoutMillis())) {
						try {
							// 若已超时 进行消息的拉取
							this.brokerController.getPullMessageProcessor().excuteRequestWhenWakeup(request.getClientChannel(), request.getRequestCommand());
						} catch (RemotingCommandException e) {
							log.error("", e);
						}
						// 继续遍历下一个PullRequest对象
						continue;
					}

					// 当前不满足要求，重新放回Hold列表中
					/**
					 * 若未超时则将该PullRequest对象存入临时的PullRequest列表中，该列表中的PullRequest对象是数据未到达但是也未超时的请求，
					 * 然后继续遍历下一个PullRequest对象直到PullRequest集合遍历完为止；
					 */
					replayList.add(request);
				}

				if (!replayList.isEmpty()) {
					// 将临时的PullRequest列表重新放入PullRequestHoldService.pullRequestTable变量中，等待下一次的遍历。
					mpr.addPullRequest(replayList);
				}
			}
		}
	}

	@Override
	public void run() {
		log.info(this.getServiceName() + " service started");
		while (!this.isStoped()) {
			try {
				this.waitForRunning(1000);
				this.checkHoldRequest();
			} catch (Exception e) {
				log.warn(this.getServiceName() + " service has exception. ", e);
			}
		}

		log.info(this.getServiceName() + " service end");
	}

	@Override
	public String getServiceName() {
		return PullRequestHoldService.class.getSimpleName();
	}
}
